package com.gss.plugin.kafka.config;

import com.gss.plugin.kafka.handle.ConsumerHandle;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

/**
 * @author: DYang
 * @since: 2023/8/24
 * @description: 消费者处理
 */
@Slf4j
@Component
public class ConsumerHandler implements CommandLineRunner {

    @Autowired
    private KafkaProperties kafkaProperties;
    @Autowired
    private KafkaConsumer<String,String> kafkaConsumer;
    @Autowired
    private ConsumerHandle consumerHandle;

    /**
     * 启动线程，使用订阅模式读取数据
     * @param args incoming main method arguments
     * @throws Exception
     */
    @Override
    public void run(String... args) throws Exception {
        boolean subscribeEnabled = kafkaProperties.getConsumer().isSubscribeEnabled();
        if (!subscribeEnabled) {
            return;
        }
        try {
            String topics = kafkaProperties.getConsumer().getTopics();
            //消费者订阅主题
            kafkaConsumer.subscribe(Arrays.asList(topics.split(",")));
            new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(5000));
                        for (ConsumerRecord<String, String> record : records) {
                            boolean handle = consumerHandle.handle(record);
                            if (handle) {
                                //同步提交，当前线程会阻塞直到offset提交成功
                                kafkaConsumer.commitSync();
                            }
                        }
                    }
                }
            }).start();
        } catch (Exception e){
            e.printStackTrace();
            log.error("【kafka 消费】 订阅模式消费数据异常",e);
        } finally{
            kafkaConsumer.close();
        }
    }


    /**
     * 消费缓存到kafka中的布控任务结果
     * @param records
     * @param consumer
     */
    @ConditionalOnProperty(name = "stars.kafka.consumer.listenerEnable",havingValue = "true")
    @KafkaListener(topics = "#{&rsquo;${stars.kafka.consumer.topics}&rsquo;.split(&rsquo;,&rsquo;)}",containerFactory = "dispositionContainerFactory")
    public void consumerListener(List<ConsumerRecord<String, String>> records, KafkaConsumer<String,String> consumer) {
        try {
            for (ConsumerRecord<String, String> record : records) {
                boolean handle = consumerHandle.handle(record);
                if (handle) {
                    //同步提交，当前线程会阻塞直到offset提交成功
                    consumer.commitSync();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("【kafka 消费】 监听模式消费异常 ",e);
        }
    }

}
